package com.pisolution.scheduler.runtime.master.scheduler;

import com.pisolution.scheduler.core.tuple.Tuple2;
import com.pisolution.scheduler.core.tuple.Tuple3;
import com.pisolution.scheduler.core.utils.JsonUtils;
import com.pisolution.scheduler.core.utils.Stopper;
import com.pisolution.scheduler.datasource.alert.AlertService;
import com.pisolution.scheduler.datasource.instance.ExecutionState;
import com.pisolution.scheduler.datasource.instance.TaskInstance;
import com.pisolution.scheduler.datasource.instance.WorkflowInstance;
import com.pisolution.scheduler.remote.utils.Host;
import com.pisolution.scheduler.runtime.SchedulerContext;
import com.pisolution.scheduler.runtime.master.MasterConfig;
import com.pisolution.scheduler.runtime.master.cache.TaskInstanceCache;
import com.pisolution.scheduler.runtime.master.cache.TaskInstanceCacheManager;
import com.pisolution.scheduler.runtime.master.dispatcher.executor.ExecutorManager;
import com.pisolution.scheduler.runtime.master.dispatcher.executor.NettyExecutorManager;
import com.pisolution.scheduler.runtime.process.ProcessService;
import com.pisolution.scheduler.runtime.queue.TaskQueue;
import com.pisolution.scheduler.runtime.queue.TaskQueueService;
import com.pisolution.scheduler.runtime.workflow.TaskNode;
import com.pisolution.scheduler.runtime.workflow.Timeout;
import com.pisolution.scheduler.runtime.workflow.TimeoutStrategy;

import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.annotation.Nullable;
import java.util.Date;
import java.util.Objects;
import java.util.concurrent.Callable;

/**
 * 负责任务实例的存储与转交执行.
 *
 * @author sora
 * @version 0.1.0
 */
public final class TaskScheduler implements Callable<Boolean> {
    private static final Logger LOGGER = LoggerFactory.getLogger(TaskScheduler.class);

    // -------------------------------------------------------------------------------------------------

    /** 所属工作流实例. */
    private WorkflowInstance workflow;

    /** 待处理任务实例. */
    private TaskInstance task;

    /** 待处理任务实例超时设置. */
    private final Timeout timeout;

    // -------------------------------------------------------------------------------------------------

    /** 用于读取任务提交次数与间隔配置. */
    private final MasterConfig config;

    /** 用于查询任务实例相关信息. */
    private final ProcessService service;

    /** 存储任务实例的队列 该队列会定期被扫描. */
    private final TaskQueue<String> queue;

    /** 用于缓存运行中的任务实例. */
    private final TaskInstanceCache cache;

    /** 用于发送告警信息. */
    private final AlertService alertService;

    /** 用于发送控制指令控制任务的执行. */
    private final ExecutorManager<Boolean> executorManager;

    // -------------------------------------------------------------------------------------------------

    /** 是否需要取消当前任务的执行. */
    private boolean isNeedCancel;

    /** 是否需要检测当前任务是否超时. */
    private boolean isNeedCheckTaskTimedOut;

    /** 任务是否已经取消执行. */
    private boolean isTaskKilled;

    // -------------------------------------------------------------------------------------------------
    //  初始化
    // -------------------------------------------------------------------------------------------------

    /**
     * 初始化 {@link TaskScheduler}.
     *
     * @param task 待处理任务实例.
     */
    public TaskScheduler(final TaskInstance task) {
        this.task = task;
        this.timeout = ((TaskNode) JsonUtils.deserialize(task.getContent(), TaskNode.class)).getTimeout();

        this.config = SchedulerContext.get(MasterConfig.class);
        this.service = SchedulerContext.get(ProcessService.class);
        this.queue = SchedulerContext.get(TaskQueueService.class);
        this.cache = SchedulerContext.get(TaskInstanceCacheManager.class);
        this.alertService = SchedulerContext.get(AlertService.class);
        this.executorManager = SchedulerContext.get(NettyExecutorManager.class);

        this.isNeedCancel = false;
        this.isNeedCheckTaskTimedOut = this.timeout != null;
        this.isTaskKilled = false;
    }

    // -------------------------------------------------------------------------------------------------
    //  属性访问
    // -------------------------------------------------------------------------------------------------

    public TaskInstance getTask() {
        return this.task;
    }

    // -------------------------------------------------------------------------------------------------
    //  任务实例调度
    // -------------------------------------------------------------------------------------------------

    /**
     * 取消当前任务实例的执行.
     */
    public void kill() {
        this.isNeedCancel = true;
    }

    @Override
    public Boolean call() {
        // 1. 查询当前任务实例所属工作流实例.
        this.workflow = this.service.findWorkflowInstanceById(this.task.getWorkflowInstanceId());

        // 2. 提交当前任务实例至数据库中并存储与共享队列中等待任务扫描器扫描并转交给工作节点处理.
        this.task = this.submit();

        // 3. 如果提交失败 则判定该任务实例执行失败.
        if (this.task == null) {
            LOGGER.error("Could not submit the task to the database and the shared queue.");

            return false;
        }

        // 4. 如果当前任务实例尚未执行完成 则等待其处理完成.
        boolean isTaskResolved = false;
        if (!this.task.getState().isFinished()) {
            isTaskResolved = this.waitTaskQuit();
        }

        // 5. 更新任务实例信息.
        this.task.setEndTime(new Date());
        this.service.saveTaskInstance(this.task);

        LOGGER.info("The execution of the task {} is completed.", this.task.getName());

        return isTaskResolved;
    }

    /**
     * 提交当前任务实例至数据库与共享队列中.
     *
     * @return 数据库中存储的当前任务实例数据.
     */
    @Nullable
    private TaskInstance submit() {
        // 1. 从管理节点配置中读取任务提交的可用次数与时间间隔.
        final int interval = this.config.getMasterTaskCommitInterval();
        final int maxRetryCount = this.config.getMasterTaskCommitRetryCount();

        // 2. 任务提交成功前不断尝试提交任务.
        int retryCount = 1;
        TaskInstance task = null;

        // 3. 用于判定数据库与共享队列是否已经提交成功 {数据库标识, 队列标识} (IDE不支持循环体中变量值变更检测 故以此方式绕过其警告信息).
        Tuple2<Boolean, Boolean> flag = new Tuple2<>(false, false);

        while (retryCount <= maxRetryCount) {
            try {
                // 4. 如果尚未成功提交到数据库中 则尝试提交至数据库中.
                if (!flag.field1) {
                    task = this.service.submitTask(this.task);
                    flag.field1 = task != null && task.getId() != 0;
                }

                // 5. 如果成功提交至数据库中且尚未提交至共享队列中 则尝试提交至共享队列中.
                if (flag.field1 && !flag.field2) {
                    flag.field2 = this.dispatchTask(task);
                }

                // 6. 如果已经成功提交至数据库与共享队列中 则判定为提交成功.
                if (flag.field1 && flag.field2) {
                    return task;
                }

                // 7. 显示相应错误日志并等待下一次重试.
                if (!flag.field1) {
                    LOGGER.error("Could not save the task {} into the database and wait for the next retry.", this.task.getName());
                }
                if (!flag.field2) {
                    LOGGER.error("Could not submit the task {} to the shared queue and wait for the next retry.", this.task.getName());
                }

                Thread.sleep(interval);
            } catch (Exception e) {
                LOGGER.error("Could not save the task into the database and submit it to the shared queue.", e);
            }

            retryCount++;
        }

        return task;
    }

    /**
     * 提交当前任务至共享队列中以等待任务扫描器扫描并转交给工作节点处理.
     *
     * @param task 待提交任务实例.
     *
     * @return 如果提交成功则返回 {@code true} 否则返回 {@code false}.
     */
    private boolean dispatchTask(final TaskInstance task) {
        try {
            // 1. 如果待提交任务实例已经执行完成 则判定为提交成功.
            if (task.getState().isFinished()) {
                LOGGER.info("The task {} submitted is already finished with state {}.", task.getName(), task.getState());

                return true;
            }

            // 2. 如果待提交任务实例尚在运行中 则判定为提交成功.
            if (task.getState() == ExecutionState.RUNNING) {
                LOGGER.info("The task {} submitted is still running.", task.getName());

                return true;
            }

            LOGGER.info("Ready to submit the task {} to the shared queue.", task.getName());

            // 3. 提交任务至共享队列中.
            this.queue.put(String.join(
                "_",
                String.valueOf(this.workflow.getPriority().getCode()),
                String.valueOf(this.workflow.getId()),
                String.valueOf(task.getPriority().getCode()),
                String.valueOf(task.getId()),
                task.getWorkerGroup()
            ));

            LOGGER.info("Succeed to submit the task {} to the shared queue.", task.getName());

            return true;
        } catch (Exception e) {
            LOGGER.error("Could not submit the task {} to the shared queue.", task.getName(), e);
        }

        return false;
    }

    /**
     * 等待任务退出执行.
     *
     * @return 如果任务成功退出执行则返回 {@code true} 否则返回 {@code false}.
     */
    @SuppressWarnings("BusyWait")
    private Boolean waitTaskQuit() {
        LOGGER.info("Wait until the task is completed.");

        // 1. 查询任务实例的最新状态.
        this.task = this.service.findTaskInstanceById(this.task.getId());

        while (Stopper.isRunning()) {
            try {
                // 2. 如果工作流实例不存在 则判定任务已经退出执行.
                if (this.workflow == null) {
                    LOGGER.error("Exit the execution of the task since the workflow instance belonged is not found.");

                    return true;
                }

                // 3. 如果用户希望取消任务执行 则通知工作节点取消当前任务的执行.
                if (this.isNeedCancel || this.workflow.getState() == ExecutionState.READY_STOP) {
                    this.cancelExecution();
                }

                // 4. 如果用户希望暂停任务的执行 则尝试暂停当前任务的执行.
                if (this.workflow.getState() == ExecutionState.READY_PAUSE) {
                    this.pauseExecution();
                }

                // 5. 如果任务实例已经执行完成 则从缓存中移除该任务实例.
                if (this.task.getState().isFinished()) {
                    this.cache.remove(this.task.getId());
                    break;
                }

                // 6. 确认任务执行是否已经超时.
                if (this.checkIfTimedOut()) {
                    this.alertTimeout();

                    this.isNeedCheckTaskTimedOut = false;
                }

                // 7. 更新任务实例与工作流实例.
                this.task = this.service.findTaskInstanceById(this.task.getId());
                this.workflow = this.service.findWorkflowInstanceById(this.workflow.getId());

                Thread.sleep(1_000L);
            } catch (Exception e) {
                LOGGER.error(e.getMessage(), e);
            }
        }

        return true;
    }

    /**
     * 取消任务的执行.
     */
    private void cancelExecution() throws Exception {
        // 1. 如果任务已经取消执行 则无需处理.
        if (this.isTaskKilled) {
            return;
        }

        // 2. 更改标识.
        this.isTaskKilled = true;

        // 3. 查询任务最新状态.
        this.task = this.service.findTaskInstanceById(this.task.getId());

        // 4. 如果任务尚未提交至共享队列中 则仅需更改任务当前的状态即可.
        if (StringUtils.isBlank(Objects.requireNonNull(this.task).getHost())) {
            this.task.setState(ExecutionState.KILL);
            this.task.setEndTime(new Date());

            this.service.saveTaskInstance(this.task);
            return;
        }

        // 5. 构建系统指令告知工作节点取消任务的执行.
        final TaskKillRequestCommand command = new TaskKillRequestCommand();

        command.setTaskId(this.task.getId());
        this.executorManager.executeDirectly(new Tuple3<>(
            Host.of(this.task.getHost()),
            command.convert(),
            this.task.getWorkerGroup()
        ));

        LOGGER.info("Request the worker server to cancel the execution of the task {}.", this.task.getName());
    }

    /**
     * 如果任务还未提交到共享队列中则暂停任务的执行.
     */
    private void pauseExecution() {
        // 1. 查询最新的任务实例对象.
        this.task = this.service.findTaskInstanceById(this.task.getId());

        // 2. 如果任务实例不存在 则无需处理.
        if (this.task == null) {
            return;
        }

        // 3. 如果任务尚未提交到共享队列中 则暂停该任务的执行.
        if (StringUtils.isBlank(this.task.getHost())) {
            this.task.setState(ExecutionState.PAUSE);
            this.task.setEndTime(new Date());

            this.service.saveTaskInstance(this.task);
        }
    }

    /**
     * 校验当前任务是否已经超时.
     *
     * @return 如果已经超时则返回 {@code true} 否则返回 {@code false}.
     */
    private boolean checkIfTimedOut() {
        // 1. 如果无需检测任务是否超时或者任务开始执行时间无法获取 则判定为未超时.
        if (!this.isNeedCheckTaskTimedOut || this.task.getStartTime() == null) {
            return false;
        }

        // 2. 如果剩余可供任务执行的时间不足 则判定为任务失败.
        return this.timeout.getDuration() * 60L <= (System.currentTimeMillis() -  this.task.getStartTime().getTime()) / 1_000;
    }

    /**
     * 发送告警信息.
     */
    private void alertTimeout() {
        // 1. 如果超时策略为超时失败 则无需处理.
        if (this.timeout.getStrategy() == TimeoutStrategy.FAILURE) {
            return;
        }

        LOGGER.warn("The execution of the task {} is timed out.", this.task.getName());

        // 2. 发送告警邮件.
        this.alertService.sendTaskTimeoutAlert(this.workflow, this.task);
    }
}
